package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 需求：实时统计不同支付类型支付总金额
 * @author xuanyu
 * @date 2025/10/28
 */
public class _03GmallPaymentTumblingJob {

	public static void main(String[] args) throws Exception{
		// 1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2.数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("db-topic")
			.setGroupId("gmall-payment-g1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);
//		stream.print("kafka");


		// 3.数据转换-transformation
		// 3-1.获取支付信息数据
		SingleOutputStreamOperator<String> stream31 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析JSON
					JSONObject jsonObject = JSON.parseObject(value).getJSONObject("source");
					// 表名称
					String table = jsonObject.getString("table");
					// 比较
					return "payment_info".equals(table);
				}
			}
		);

		// 3-2.过滤获取after字段有值数据（将after为null过滤掉）
		SingleOutputStreamOperator<String> stream32 = stream31.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析json
					Object afterValue = JSON.parseObject(value).get("after");
					// 判断
					return null != afterValue;
				}
			}
		);

		// todo 指定数据事件时间字段和水位线
		SingleOutputStreamOperator<String> stream322 = stream32.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						JSONObject jsonObject = JSON.parseObject(element).getJSONObject("after");
						// 字段值：create_time
						String createTime = jsonObject.getString("create_time");
						// 转换
						Date date = format.parse(createTime);
						// 时间戳
						return date.getTime();
					}
				})
		);

		// 3-3.提取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream33 = stream322.map(
			new MapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					JSONObject jsonObject = JSON.parseObject(value).getJSONObject("after");
					// payment_type 和 total_amount
					String paymentType = jsonObject.getString("payment_type");
					Double totalAmount = jsonObject.getDouble("total_amount");
					// 返回
					return Tuple2.of(paymentType, totalAmount);
				}
			}
		);

		// 3-4.分组
		KeyedStream<Tuple2<String, Double>, String> stream34 = stream33.keyBy(tuple -> tuple.f0);
		//stream34.sum(1) ;

		// 3-5.窗口
		WindowedStream<Tuple2<String, Double>, String, TimeWindow> stream35 = stream34.window(
			TumblingEventTimeWindows.of(Time.minutes(1))
		);

		// 3-6.聚合计算
		SingleOutputStreamOperator<String> stream36 = stream35.apply(
			new WindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				@Override
				public void apply(String paymentType,
				                  TimeWindow window,
				                  Iterable<Tuple2<String, Double>> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd()) ;
					// 计算
					double amount = 0.0 ;
					for (Tuple2<String, Double> value : input) {
						amount += value.f1;
					}
					// 输出
					String output = windowStart + "," + windowEnd + "," + paymentType + "," + amount;
					out.collect(output);
				}
			}
		);

		// 4.数据接收器-sink
		stream36.print();

		// 5.触发执行-execute
		env.execute("GmallPaymentJob") ;
	}

}
